package com.codejiwei.flink.zk;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/9/8 9:21
 */
public class FlinkZookeeperSourceTest {
    public static void main(String[] args) throws Exception {
        String zkConnectString = "192.168.102.152:22181";
        String zkPath = "/test_metadata/sensor/t1";

//        CuratorFramework client = CuratorFrameworkFactory.newClient(zkConnectString, new ExponentialBackoffRetry(1000, 3));
//        CuratorFramework client = new ZkClient(zkConnectString).getClient();
//
//
//        client.start();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new ZookeeperSource(zkConnectString, zkPath))
                .map(new MapFunction<byte[], String>() {
                    @Override
                    public String map(byte[] value) throws Exception {
                        return new String(value);
                    }
                })
                .print();

        env.execute("zookeeper source demo");
    }
}
